package com.learn.confirm;

import com.learn.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author DJJ_F
 * @version 1.0
 * @date 2025/08/16
 * @description ch04 发布确认
 */
public class ProducerConfirm {
    public static final String QUEUE_NAME = "confirm_queue";
    public static final ConcurrentSkipListMap<Long, String> map = new ConcurrentSkipListMap<>();


    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMqUtils.getChannel();
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.confirmSelect();

        // 添加监听器
        channel.addConfirmListener((deliveryTag, multiple) -> {
            if(multiple) {
                // headMap(toKey, inclusive) 方法的作用是返回原映射中 “键小于或等于 key 的部分视图
                // 因此可以批量删除
                ConcurrentNavigableMap<Long, String> headMap = map.headMap(deliveryTag);
                headMap.clear();
            } else {
                map.remove(deliveryTag);
            }
            System.out.println("成功确认：" + deliveryTag);

        }, (deliveryTag, multiple) -> {
            System.out.println("确认失败：" + deliveryTag +"\n 内容：" + map.remove(deliveryTag));
        });

        long begin = System.currentTimeMillis();
        // time: 190 ms
//        confirmOne(channel);
        // time: 30 ms
//        confirmBatch(channel);
        // time: 26 ms
        confirmAsync(channel);
        long end = System.currentTimeMillis();
        System.out.println("time: " + (end - begin) + " ms");
    }

    private static void confirmOne(Channel channel) throws Exception {
        for (int i = 0; i < 1000; i++) {
            String message = "msg" + i;
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
            // 发送单个消息立马确认
            channel.waitForConfirms();
        }
    }

    private static void confirmBatch(Channel channel) throws Exception {
        for (int i = 0; i < 1000; i++) {
            String message = "msg" + i;
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
            // 批量确认
            if ((i + 1) % 100 == 0) {
                channel.waitForConfirms();
            }
        }
    }

    private static void confirmAsync(Channel channel) throws Exception {
        for (int i = 0; i < 1000; i++) {
            String message = "msgC" + i;
            // 注意在basicPublish前就不用-1
            map.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
        }
    }
}
